package flink_p1

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, SplitStream, StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.util.Collector

object FlinkTest_07_Operator_split {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val stream: DataStream[Long] = env.generateSequence(0, 100)


    //分流算子

    val splitStream: SplitStream[Long] = stream.split(data => {
      data % 2 match {
        case 0 => List("STREAM_1")
        case 1 => List("STREAM_2")
      }
    })

    //通过select选择出目标流，进行分别处理
    //    splitStream.select("STREAM_1").print("偶数")
    //    splitStream.select("STREAM_2").print("奇数")


    // Please use side outputs instead of split/select", "deprecated since 1.8.2
    //使用侧输出流：将奇数从side output输出

    val tag = new OutputTag[Long]("ou")
    val dataStream: DataStream[Long] = stream.process(new ProcessFunction[Long, Long] {
      override def processElement(value: Long, ctx: ProcessFunction[Long, Long]#Context, out: Collector[Long]): Unit = {

        value % 2 match {
          case 0 => out.collect(value) //往主流发射
          case 1 => ctx.output(tag, value) //往测流发射
        }

      }
    })

    //获取主流数据
    dataStream.print("main stream")

    //获取测流数据
    dataStream.getSideOutput(tag).print("side stream")


    env.execute("test...")

  }
}
